package com.cmb.lk50.sink.db;

import com.cmb.lk50.pojo.AbstractLog;
import com.cmb.lk50.pojo.AccessInfo;
import com.cmb.lk50.sink.AsyncLogSink;
import com.zaxxer.hikari.HikariDataSource;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.concurrent.ExecutorService;

/**
 * Author       : 蒋俊钊 80256973
 * Date         : 2019/11/12
 * Copyright    (C) ChinaMerchantsBank
 */
public class JdbcSink extends AsyncLogSink {

    private static final Logger LOGGER = LoggerFactory.getLogger(JdbcSink.class);

    private static final String INSERT_ACCESSINFO_SQL =
      "insert into access_log (host_name, host_ip,\n" +
        "host_port, method, path,\n" +
        "invoke_ip, invoke_port, query_param,\n" +
        "status, response_content, cost_time)\n" +
        "values (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)";

    private HikariDataSource hikariDataSource;

    private StatementGenerator statementGenerator;

    public JdbcSink(HikariDataSource hikariDataSource, ExecutorService asyncExecutor, StatementGenerator statementGenerator) {
        super(asyncExecutor);
        this.hikariDataSource = hikariDataSource;
        this.statementGenerator = statementGenerator;
    }

    @Override
    public void asyncWrite(AbstractLog log) {
        try (Connection connection = hikariDataSource.getConnection()) {
            if (log instanceof AccessInfo) {
                try (PreparedStatement preparedStatement = connection.prepareStatement(INSERT_ACCESSINFO_SQL)) {
                    AccessInfo accessInfo = (AccessInfo) log;
                    preparedStatement.setString(1, accessInfo.getHostName());
                    preparedStatement.setString(2, accessInfo.getHostIp());
                    preparedStatement.setInt(3, accessInfo.getHostPort());
                    preparedStatement.setString(4, accessInfo.getMethod());
                    preparedStatement.setString(5, accessInfo.getPath());
                    preparedStatement.setString(6, accessInfo.getInvokeIp());
                    preparedStatement.setInt(7, accessInfo.getInvokePort());
                    preparedStatement.setString(8, accessInfo.getQueryParam());
                    preparedStatement.setString(9, accessInfo.getStatus());
                    preparedStatement.setString(10, accessInfo.getResponseContent());
                    preparedStatement.setLong(11, accessInfo.getCostTime());
                    preparedStatement.execute();
                }
            } else {
                try (PreparedStatement preparedStatement = statementGenerator.getPsByType(connection, log)){
                    preparedStatement.execute();
                }
            }

        } catch (RuntimeException | SQLException e) {
            if (LOGGER.isDebugEnabled()) {
                LOGGER.debug("[JDBC Sink] 写入日志异常，[{}]", e.getMessage());
            }
        }
    }
}